#pragma once
#include <muduo/net/TcpServer.h>
#include <muduo/net/EventLoop.h>
#include <muduo/net/TcpConnection.h>
#include <muduo/net/Buffer.h>
#include <muduo/base/CountDownLatch.h>
#include <muduo/net/EventLoopThread.h>
#include <muduo/net/TcpClient.h>
#include "detail.hpp"
#include "fields.hpp"
#include "abstract.hpp"
#include "message.hpp"
#include <mutex>
#include <unordered_map>

using namespace bitrpc;

class MuduoBuffer : public BaseBuffer
{
public:
    using ptr = std::shared_ptr<MuduoBuffer>;

    MuduoBuffer(muduo::net::Buffer *buf)
        :_buf(buf)
    {}

//连接 mudou库
//实现 功能

    virtual size_t readableSize() override { return _buf->readableBytes(); }

    //muduo库是一个网络库，从缓冲区取出一个4字节整形，会进行网络字节序的转换
    virtual int32_t peekInt32() override { return _buf->peekInt32(); }

    virtual void retrieveInt32() override { _buf->retrieveInt32(); }

    virtual int32_t readInt32() override { return _buf->readInt32(); }

    virtual std::string retrieveAsString(size_t len) override { return _buf->retrieveAsString(len); }

private:
    muduo::net::Buffer *_buf;

};

/////////////工厂
class BufferFactory
{
public:
    template <typename... Args>
    
    //！！！！！！！！右值引用 完美转发
    static BaseBuffer::ptr create(Args &&...args)
    {
        return std::make_shared<MuduoBuffer>(std::forward<Args>(args)...);
    }
};




///////////////////////////////////////////////////////////////////
class LVProtocol : public BaseProtocol
{
public:
    using ptr = std::shared_ptr<LVProtocol>;

    // |--Len--|--VALUE--|
    // |--Len--|--mtype--|--idlen--|--id--|--body--|
    //判断缓冲区中的数据量是否足够一条消息的处理

//缓冲区中 读取消息
    virtual bool canProcessed(const BaseBuffer::ptr &buf) override
    {
        if(buf->readableSize()<lenFieldsLength){
            //不够 一条消息的处理
            return false;
        }

        int32_t total_len=buf->peekInt32();
        if(buf->readableSize()<(total_len+lenFieldsLength))
        {
            return false;
        }
        //确保 有一条 完整的lv 协议
        return true;
    }

    //////////////////////////////////////////////////////

//缓冲区中取出 反序列化 消息
    virtual bool onMessage(const BaseBuffer::ptr &buf, BaseMessage::ptr &msg)
    {
        //!! 当 使用onMessage的时候，默认认为 缓冲区的数据 为一条足够的完整消息
        int32_t total_len = buf->readInt32();  // 读取总⻓度
        MType mtype = (MType)buf->readInt32(); // 读取数据类型
        int32_t id_len = buf->readInt32();      // 读取id⻓度
        int32_t body_len = total_len - id_len - idlenFieldsLength - mtypeFieldsLength;

        std::string id=buf->retrieveAsString(id_len);
        std::string body=buf->retrieveAsString(body_len);
        
/*
enum class MType {
      
        //request
        //response
        
        REQ_RPC = 0,
        RSP_RPC,
        REQ_TOPIC,
        RSP_TOPIC,
        REQ_SERVICE,
        RSP_SERVICE
    }
*/
        msg=MessageFactory::create(mtype);
        if(msg.get()==nullptr)
        {
            ELOG("消息类型错误，构造 消息对象失败");
            return false;
        }

        bool ret=msg->unserialize(body);
        //!!!!!!!!!!! see abstract 中设计

        if(ret == false)
        {
            ELOG("消息正文反序列化失败！");
            return false;
        }

        msg->SetId(id);
        msg->SetMType(mtype);
        return true;
    }


//！！！！！！！！序列化
    virtual std::string serialize(const BaseMessage::ptr &msg) override
    {
        // |--Len--|--mtype--|--idlen--|--id--|--body--|
        std::string body = msg->serialize();
        std::string id = msg->GetId();
        auto mtype = htonl((int32_t)msg->GetMType());
        int32_t id_len = htonl(id.size());
        int32_t h_total_len = mtypeFieldsLength + idlenFieldsLength + id.size() + body.size();
        int32_t n_total_len = htonl(h_total_len);

        std::string result;
        result.reserve(h_total_len);// 将空间 提前分配好
        result.append((char*)&n_total_len, lenFieldsLength);
        result.append((char*)&mtype, mtypeFieldsLength);
        result.append((char*)&id_len, idlenFieldsLength);
        result.append(id);
        result.append(body);

        return result;
    }

//！！！！！！！！避免魔幻字符的出现 
//不要直接使用常数
private:
    const size_t lenFieldsLength = 4;
    const size_t mtypeFieldsLength = 4;
    const size_t idlenFieldsLength = 4;

};

class ProtocolFactory
{
public:
    template <typename... Args>
    static BaseProtocol::ptr create(Args &&...args)
    {
        return std::make_shared<LVProtocol>(std::forward<Args>(args)...);
    }
};


///////////////////////////////////////////////////////////////////////////////////

class MuduoConnection : public BaseConnection
{
public:
    using ptr = std::shared_ptr<MuduoConnection>;

    MuduoConnection(BaseProtocol::ptr& protocol, const muduo::net::TcpConnectionPtr &conn)
        :_protocol(protocol)
        ,_conn(conn)
    {}

    virtual void send(const BaseMessage::ptr &msg) override
    {
        std::string body = _protocol->serialize(msg);
        _conn->send(body);
    }

    virtual void shutdown() override
    {
        _conn->shutdown();
    }

    virtual bool connected() override
    {
        return _conn->connected();
    }

private:
    BaseProtocol::ptr _protocol;
    muduo::net::TcpConnectionPtr _conn;

};

class ConnectionFactory
{
public:
    template <typename... Args>
    static BaseConnection::ptr create(Args &&...args)
    {
        return std::make_shared<MuduoConnection>(std::forward<Args>(args)...);
    }
};

////////////////////////////////////////////////////////////////////////////////
class MudouServer:public BaseServer{
    public:
        using ptr=std::shared_ptr<MudouServer>;
        MudouServer(int port):
        _server(&_baseloop,muduo::net::InetAddress("0.0.0.0",port),
            "MudouServer",muduo::net::TcpServer::kReusePort),
             _protocol(ProtocolFactory::create()){}
        
        //
        virtual void start() {
                _server.setConnectionCallback(std::bind(&MuduoServer::onConnection, this, std::placeholders::_1));
                _server.setMessageCallback(std::bind(&MuduoServer::onMessage, this, 
                    std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
                _server.start();//先开始监听
                _baseloop.loop();//开始死循环事件监控
            }
    private:
        void onConnection(const muduo::net::TcpConnectionPtr &conn){

            if(conn->connected()){
                std::cout<<"连接 建立\n";
                auto muduo_conn=ConnectionFactory::create(conn,_protocol);
                {
                    //加锁 给工厂 进行连接
                    //！！！unique_lock
                    std::unique_lock<std::mutex> lock(_mutex);
                    _conns.insert(std::make_pair(conn,muduo_conn));//建立 连接
                }
                if(_cb_connection) _cb_connection(muduo_conn);
            }
            else{
                std::cout<<"连接 断开\n";
                BaseConnection::ptr muduo_conn;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it=_conns.find(conn);
                    if(it==_conns.end()){
                        return;
                    }
                    muduo_conn=it->second;
                    _conns.erase(conn);
                }
                if(_cb_close)
                    _cb_close(muduo_conn);
            }
        }

        void onMessage(const muduo::net::TcpConnectionPtr &conn,muduo::net::Buffer *buf,muduo::Timestamp)
        {
            DLOG("连接有 数据到来，开始处理！");
            auto base_buf=BufferFactory::create(buf);
            while(1)
            {
                if(_protocol->canProcessed(base_buf)==false)
                {
                    //数据不足
                    if(base_buf->readableSize()>maxDataSize)
                    {
                        conn->shutdown();
                        ELOG("缓冲区 数据过大！");
                        return;
                    }
                    break;
                }

                BaseMessage::ptr msg;
                bool ret=_protocol->onMessage(base_buf,msg);
                if(ret==false)
                {
                    conn->shutdown();
                    ELOG("缓冲区 数据错误");
                    return;
                }

                BaseConnection::ptr base_conn;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto iter=_conns.find(conn);
                    if(iter==_conns.end())
                    {
                        conn->shutdown();
                        return;
                    }

                    base_conn=iter->second;
                }
                if(_cb_message)
                {
                    _cb_message(base_conn,msg);
                }
            }
        }

    private:
        const size_t maxDataSize = (1 << 16);
        BaseProtocol::ptr _protocol;//协议

        muduo::net::EventLoop _baseloop;//
        muduo::net::TcpServer _server;//调用mudou库接口

        std::mutex _mutex;//加锁
        std::unordered_map<muduo::net::TcpConnectionPtr, BaseConnection::ptr> _conns;
        //建立映射
};



////////////////////////////////////////////////////////////////
class MuduoClient : public BaseClient
{
public:
    using ptr = std::shared_ptr<MuduoClient>;

    MuduoClient(const std::string &ip, int port)
        :_protocol(ProtocolFactory::create())
        ,_baseloop(_loopthread.startLoop())
        ,_downlatch(1)
        ,_client(_baseloop, muduo::net::InetAddress(ip, port), "DictClient")
    {}

    virtual void connect() override
    {
        DLOG("设置回调函数，连接服务器");
        _client.setConnectionCallback(std::bind(&MuduoClient::onConnection, this, std::placeholders::_1));
        _client.setMessageCallback(std::bind(&MuduoClient::onMessage, this, std::placeholders::_1, 
                                   std::placeholders::_2, std::placeholders::_3));
        
        //连接服务器
        _client.connect();
        _downlatch.wait();

        DLOG("连接服务器成功！");
    }

    virtual void shutdown() override { return _client.disconnect(); }

    virtual bool send(const BaseMessage::ptr &msg) override
    {
        if(_conn->connected() == false)
        {
            ELOG("连接已断开！");
            return false;
        }
        else
        {
            _conn->send(msg);
        }

        return true;
    }

    virtual BaseConnection::ptr connection() override { return _conn; }

    virtual bool connected() override { return (_conn && _conn->connected()); }

private:
    void onConnection(const muduo::net::TcpConnectionPtr &conn)
    {
        if(conn->connected())
        {
            std::cout << "连接建立！\n";
            _downlatch.countDown();//计数--，为0时唤醒阻塞
            _conn = ConnectionFactory::create(_protocol, conn);
        }
        else
        {
            std::cout << "连接断开！\n";
            _conn.reset();
        }
    }

    void onMessage(const muduo::net::TcpConnectionPtr &conn, muduo::net::Buffer *buf, muduo::Timestamp)
    {
        DLOG("连接有数据到来，开始处理！");
        auto base_buf = BufferFactory::create(buf);
        //！！！！！！！！！！！！！
        //这个无限循环的作用是持续处理从网络连接接收到的数据
        //直到没有足够的数据可供处理或发生错误为止
        while(1)
        {
            if(_protocol->canProcessed(base_buf) == false)
            {
                //数据不足
                if(base_buf->readableSize() > maxDataSize)
                {
                    conn->shutdown();
                    ELOG("缓冲区中数据过大！");
                    return;
                }

                DLOG("数据量不足！");
                break;
            }

            BaseMessage::ptr msg;
            bool ret = _protocol->onMessage(base_buf, msg);
            if(ret == false)
            {
                conn->shutdown();
                ELOG("缓冲区中数据错误！");
                return;
            }

            if(_cb_message)
            {
                _cb_message(_conn, msg);
            }
        }
    }

private:
    const size_t maxDataSize = (1 << 16);
    BaseProtocol::ptr _protocol;
    BaseConnection::ptr _conn;
    muduo::CountDownLatch _downlatch;
    muduo::net::EventLoopThread _loopthread;
    muduo::net::EventLoop *_baseloop;
    muduo::net::TcpClient _client;

};

class ClientFactory
{
public:
    template <typename... Args>
    static BaseClient::ptr create(Args &&...args)
    {
        return std::make_shared<MuduoClient>(std::forward<Args>(args)...);
    }
};
